package com.quectel.util.client.snowflake;

import com.google.common.base.Preconditions;
import com.quectel.util.ex.kit.SnowflakeException;
import com.quectel.util.thread.ThreadPoolUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import java.security.SecureRandom;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * <pre>
 * 雪花算法客户端 需要依赖Redis来进行机器号的申请
 * 雪花算法解析 结构 Snowflake的结构如下(每部分用-分开):
 * 0 - 00000000000000000000000000000000000000000 - 0000000000 - 000000000000
 * 第一位为未使用 表示正数
 * 接下来的41位为毫秒级时间(41位的长度可以使用69年)
 * 然后是10位的长度最多支持部署1024个节点 0~1023
 * 最后12位是毫秒内的计数（12位的计数顺序号支持每个节点每毫秒产生4096个ID序号）
 * 一共加起来刚好64位，为一个Long型。
 * </pre>
 *
 * @author: caolu
 * @email: louis.cao@quectel.com
 * @date: 2023-04-25 19:03:00
 */
@Slf4j
public class SnowflakeClient {
    /**
     * 起始的时间戳 2018-01-01 00:00:00.000
     */
    private static final long START_STAMP = 1514736000000L;

    /**
     * 时间差所占用的bit
     */
    private static final byte TIME_BIT = 41;

    /**
     * 机器号所占的bit
     */
    private static final byte MACHINE_ID_BIT = 10;

    /**
     * 序列所占用的长度
     */
    private static final byte SEQ_BIT = 12;

    /**
     * 机器号最大值 0~1023
     */
    private static final int MACHINE_ID_MAX = ~((-1) << MACHINE_ID_BIT);

    /**
     * 序列的最大值 0~4095
     */
    private static final int SEQ_MAX = ~((-1) << SEQ_BIT);

    /**
     * 机器标识 0~1023
     */
    private int machineId = -1;
    /**
     * 序列号 0~4095
     */
    private int sequence = 0;
    /**
     * 上一次时间戳
     */
    private long lastStamp = -1L;


    private StringRedisTemplate redisTemplate;

    /**
     * 项目名称
     */
    private String projectName;

    /**
     * 标记客户端id
     */
    private final String clientId = UUID.randomUUID().toString();

    /**
     * redis加锁 按照项目名称和机器号 t1:项目名 t2:机器号
     */
    private static final Function<Tuple2<String, Integer>, String> REDIS_LOCK_FUNC =
            tuple2 -> String.format("Snowflake:%s:%d", tuple2.getT1(), tuple2.getT2());

    /**
     * 锁占用时长 默认3分钟
     */
    private static final int LOCK_EXPIRED_MINUTES = 3;

    /**
     * 上报间隔
     */
    private static final int REPORT_DURATION_SECONDS = 10;

    /**
     * 定时任务
     */
    private final ScheduledThreadPoolExecutor executor = ThreadPoolUtils.createSchedulePool(1, "Snowflake-heartbeat-task-pool-");

    /**
     * 无参构造
     */
    private SnowflakeClient() {
    }

    /**
     * 创建雪花算法客户端
     *
     * @param projectName
     * @param redisTemplate
     */
    public SnowflakeClient(String projectName, StringRedisTemplate redisTemplate) {

        if (projectName == null || redisTemplate == null) {
            throw new NullPointerException();
        }
        this.projectName = projectName;
        this.redisTemplate = redisTemplate;

        // 第一次分配
        this.machineId = obtainMachineId();

        // 启动定时
        this.startTask();
    }

    private void startTask() {
        executor.scheduleAtFixedRate(() -> {
            try {
                //有效期为3分钟 如果3分钟内没有上报心跳那么这个虚拟机器id就会被收回 我们在这里10秒报一次
                String lockId = REDIS_LOCK_FUNC.apply(Tuples.of(this.projectName, this.machineId));
                String redisCacheClientId = redisTemplate.opsForValue().get(lockId);

                if (Objects.equals(redisCacheClientId, this.clientId)) {
                    //如果是我存的机器号那就重新设置有效期(存在当前机器没有挂掉等待3分钟后继续上报心跳，在这个过程中别的机器可能占用了当前机器的id 这种情况不能重新设置)
                    redisTemplate.opsForValue().set(lockId, this.clientId, LOCK_EXPIRED_MINUTES, TimeUnit.MINUTES);
                } else {
                    //重新申请 线程池阻塞一直等待allocMachineId执行成功成功为止
                    this.machineId = obtainMachineId();
                }
            } catch (Exception e) {
                log.error("Snowflake alloc machine id throw exception", e);
            }
        }, 0, REPORT_DURATION_SECONDS, TimeUnit.SECONDS);
    }


    private int obtainMachineId() {

        for (int i = 0; i <= MACHINE_ID_MAX; i++) {
            try {
                Boolean lockSuccess = redisTemplate.opsForValue().setIfAbsent(REDIS_LOCK_FUNC.apply(Tuples.of(this.projectName, i)),
                        this.clientId, LOCK_EXPIRED_MINUTES, TimeUnit.MINUTES);
                if (Boolean.TRUE.equals(lockSuccess)) {
                    return i;
                }
            } catch (Exception e) {
                log.error("Snowflake apply machine id request failed", e);
            }
        }
        throw new SnowflakeException("Snowflake alloc machine id not success");
    }

    /**
     * 回收机器id和心跳任务
     */
    public void free() {
        if (isReady()) {
            //不需要删除当前的占用（分布式的时间可能有误差 立马释放会造成另一台机器占用，这两台时间如果相差几秒就会产生重复id
            //当然不立马释放方案也解决不了时间相差很多，比如时间误差超过1天重新占用以前的就会有问题，所以集群时间误差不要超过3分钟即机器号锁定时间）
        }
        ThreadPoolUtils.shutdown(executor);
    }

    /**
     * 是否申请到了机器id
     *
     * @return
     */
    public boolean isReady() {
        return this.machineId >= 0;
    }


    /**
     * 高性能产生分布式不重复id
     *
     * @return 23位的不重复订单号
     */
    public String getOrderNo() {
        Preconditions.checkArgument(this.isReady());
        long nextId = this.nextId();
        SnowflakeClient.IdParseInfo parse = parse(nextId);
        return new SimpleDateFormat("yyMMddHHmmssSSS").format(parse.getCreateTime()) +
                String.format("%04d", parse.getMachineId()) +
                String.format("%04d", parse.getSequence());
    }

    /**
     * 具有前缀的高性能产生分布式不重复id
     *
     * @return prefix+23位的不重复订单号
     */
    public String getOrderNo(String prefix) {
        return prefix + getOrderNo();
    }

    /**
     * 产生下一个ID
     *
     * @return
     */
    public long nextId() {
        if (!isReady()) {
            throw new SnowflakeException("Snowflake id not acquire success");
        }
        synchronized (this) {
            long currStamp = getNewStamp();
            while (currStamp < lastStamp) {
                if (lastStamp - currStamp > 20L * 1000) {
                    //时钟回拨20秒以上
                    throw new SnowflakeException("Clock back too long");
                }
                //等待时间的二倍
                try {
                    this.wait((lastStamp - currStamp) << 1);
                } catch (InterruptedException e) {
                    log.error("Snowflake thread sleep error", e);
                    // Restore interrupted state...
                    Thread.currentThread().interrupt();
                }

                currStamp = getNewStamp();
                if (currStamp < lastStamp) {
                    throw new SnowflakeException("The time is still wrong after waiting");
                }
            }


            if (currStamp == lastStamp) {
                //相同毫秒内，序列号自增
                sequence = (sequence + 1) & SEQ_MAX;
                //同一毫秒的序列数已经达到最大
                if (sequence == 0L) {
                    currStamp = getNextMill();
                }
            } else {
                //不同毫秒内，序列号置为0  低并发情况下这个值会大多数情况为0 如果采取分片建议去0~100随机值
                sequence = new SecureRandom().nextInt(100);
            }

            lastStamp = currStamp;

            return  //机器号部分
                    (currStamp - START_STAMP) << (SEQ_BIT + MACHINE_ID_BIT)
                            //机器标识部分
                            | (long) machineId << SEQ_BIT
                            //序列号部分
                            | sequence;
        }
    }


    private long getNextMill() {
        long mill = getNewStamp();
        while (mill <= lastStamp) {
            mill = getNewStamp();
        }
        return mill;
    }

    private static long getNewStamp() {
        return System.currentTimeMillis();
    }

    /**
     * id反向解析
     *
     * @return
     */
    public static SnowflakeClient.IdParseInfo parse(long id) {
        SnowflakeClient.IdParseInfo idParseInfo = new SnowflakeClient.IdParseInfo();
        // 由于时间在最高位直接右移动到底即可不需要与最大值 再加上其实时间
        idParseInfo.setCreateTime(new Date((id >> (SEQ_BIT + MACHINE_ID_BIT)) + START_STAMP));
        // 右移后最高位可能有值所以在与上最大值
        idParseInfo.setMachineId((int) (id >> SEQ_BIT) & MACHINE_ID_MAX);
        idParseInfo.setSequence((int) (id & SEQ_MAX));
        return idParseInfo;
    }

    /**
     * id解析
     */
    @Data
    public static class IdParseInfo {
        /**
         * 这个id产生的时间
         */
        private Date createTime;
        /**
         * 机器id
         */
        private int machineId;
        /**
         * 序列号
         */
        private int sequence;

    }
}
